package org.study.hadoop.md.ch5;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.study.hadoop.md.util.MRDPUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Map;

/**
 * reduce端连接：给定一个用户信息集合和一个用户评论列表，通过为每一条评论添加创建该评论的用户信息来丰富评论的内容
 */
public class ReduceSideJoinDriver extends Configured implements Tool {

    public static class UserJoinReducer extends Reducer<Text, Text, Text, Text> {
        private ArrayList<Text> listA = new ArrayList<Text>();
        private ArrayList<Text> listB = new ArrayList<Text>();
        private String joinType = null;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            joinType = context.getConfiguration().get("join.type");
        }

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            listA.clear();
            listB.clear();
            for (Text text : values) {
                if (text.charAt(0) == 'A') {
                    listA.add(new Text(text.toString().substring(1)));
                } else if (text.charAt(0) == 'B') {
                    listB.add(new Text(text.toString().substring(1)));
                }
            }
            executeJoinLogic(context);
        }

        private void executeJoinLogic(Context context) throws IOException, InterruptedException {
            if (joinType.equalsIgnoreCase("inner")) {
                if (!listA.isEmpty() && !listB.isEmpty()) {
                    for (Text A : listA) {
                        for (Text B : listB) {
                            context.write(A, B);
                        }
                    }
                }
            } else if (joinType.equalsIgnoreCase("left_outer")) {
                for (Text A : listA) {
                    if (!listB.isEmpty()) {
                        for (Text B : listB) {
                            context.write(A, B);
                        }
                    } else {
                        context.write(A, new Text("NULL"));
                    }
                }
            } else if (joinType.equalsIgnoreCase("right_outer")) {
                for (Text B : listB) {
                    if (!listA.isEmpty()) {
                        for (Text A : listA) {
                            context.write(A, B);
                        }
                    } else {
                        context.write(new Text("NULL"), B);
                    }
                }
            } else if (joinType.equalsIgnoreCase("full_outer")) {
                if (!listA.isEmpty()) {
                    for (Text A : listA) {
                        if (!listB.isEmpty()) {
                            for (Text B : listB) {
                                context.write(A, B);
                            }
                        } else {
                            context.write(A, new Text("NULL"));
                        }
                    }
                } else {
                    for (Text B : listB) {
                        context.write(new Text("NULL"), B);
                    }
                }
            } else if (joinType.equalsIgnoreCase("anti")) {
                if (listA.isEmpty() ^ listB.isEmpty()) {
                    for (Text A : listA) {
                        context.write(A, new Text("NULL"));
                    }
                    for (Text B : listB) {
                        context.write(new Text("NULL"), B);
                    }
                }
            } else {
                throw new RuntimeException("Join type not set to inner, left_outer, right_outer, full_outer, anti");
            }
        }
    }

    public static class CommentJoinMapper extends Mapper<Object, Text, Text, Text> {
        private Text outKey = new Text();
        private Text outValue = new Text();

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
            String userId = parsed.get("UserId");
            if (userId == null) {
                return;
            }
            outKey.set(userId);
            outValue.set("B" + value.toString());
            context.write(outKey, outValue);
        }
    }

    public static class UserJoinMapper extends Mapper<Object, Text, Text, Text> {
        private Text outKey = new Text();
        private Text outValue = new Text();

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
            String userId = parsed.get("Id");
            if (userId == null) {
                return;
            }
            outKey.set(userId);
            outValue.set("A" + value.toString());
            context.write(outKey, outValue);
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Path outPath = new Path("out");
        Path inPath1 = new Path("user.txt");
        Path inPath2 = new Path("comment.txt");
        FileSystem fs = outPath.getFileSystem(getConf());
        if (fs.exists(outPath)) {
            fs.delete(outPath, true);
        }
        String joinType = "left_outer";
        Job job = Job.getInstance(getConf(), "Reduce Side Join");
        job.getConfiguration().set("join.type", joinType);
        job.setJarByClass(ReduceSideJoinDriver.class);
        MultipleInputs.addInputPath(job, inPath1, TextInputFormat.class, UserJoinMapper.class);
        MultipleInputs.addInputPath(job, inPath2, TextInputFormat.class, CommentJoinMapper.class);
        job.setReducerClass(UserJoinReducer.class);
        FileOutputFormat.setOutputPath(job, outPath);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new ReduceSideJoinDriver(), args);
        System.exit(res);
    }
}
